package com.navinfo.platform.flink.cdc.streaming.functions;

import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

/**
 * 负责将CDC原始的SourceRecord消息,保存成一个json字符串
 * <p>
 * CDC的Streaming API,功能非常有限
 * 对于每个消息,它默认使用StringDebeziumDeserializationSchema类来转成String类型
 * 而此类转的String,是用StringBuilder拼的,格式很随意,后续步骤很难解析,所以创建了本类
 */
public class JsonDebeziumDeserializationSchema implements DebeziumDeserializationSchema<String> {
    private static final long serialVersionUID = 1168848948270918233L;

    public JsonDebeziumDeserializationSchema() {
    }

    /**
     * @param sourceRecord SourceRecord是kafka中的类,它里面有用的信息是其value属性
     *                     value属性是一个org.apache.kafka.connect.data.Struct类型的
     */
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        //只要sourceRecord的value字段,是个Struct类型的,将此struct转成json即可
        Struct struct = (Struct) (sourceRecord.value());
        JSONObject message = new JSONObject();
        //获取op字段,增:c,删:d,改:u
        message.put("op", struct.getString("op"));
        //获取ts_ms字段,时间戳
        message.put("ts_ms", struct.getInt64("ts_ms"));
        //获取before字段(新增操作无此字段)
        addNode(message, struct.getStruct("before"), "before");
        //获取after字段(删除操作无此字段)
        addNode(message, struct.getStruct("after"), "after");
        //获取source字段,里面有库名,表名等信息
        addNode(message, struct.getStruct("source"), "source");
        collector.collect(message.toJSONString());
    }

    private void addNode(JSONObject message, Struct struct, String nodeName) {
        if (struct != null) {
            JSONObject node = new JSONObject();
            message.put(nodeName, node);
            Schema bSchema = struct.schema();
            for (Field field : bSchema.fields()) {
                node.put(field.name().toLowerCase(), struct.get(field));
            }
        }
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}
